package edu.zju.gis.dldsj.utils;

import edu.zju.gis.dldsj.type.*;
import edu.zju.gis.dldsj.type.Row;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

/**
 * @author yanlo yanlong_lee@qq.com
 * @version 1.0 2018/07/16
 */
public final class HBaseHelper implements Closeable {
    private Connection connection;
    private Admin admin;

    private static class InstanceHolder {
        static HBaseHelper instance = new HBaseHelper(HBaseConfiguration.create());
    }

    public static HBaseHelper getInstance() {
        return InstanceHolder.instance;
    }

    /**
     * @param conf HBase配置，一般把$HADOOP_HOME/etc/hadoop/core-site.xml
     *             和$HBASE_HOME/conf/hbase-site.xml放在classpath下，通过
     *             {@link HBaseConfiguration#create()}方法生成
     */
    private HBaseHelper(Configuration conf) {
        try {
            this.connection = ConnectionFactory.createConnection(conf);
            this.admin = connection.getAdmin();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void close() throws IOException {
        admin.close();
    }

    /**
     * 读取表的存在状态
     *
     * @param tableName 表名
     * @return true-已存在
     */
    public boolean exists(String tableName) throws IOException {
        return admin.tableExists(TableName.valueOf(tableName));
    }

    /**
     * 创建表
     *
     * @param tableName 表名
     * @param families  列族列表
     */
    public void createTable(String tableName, String... families) throws IOException {
        HTableDescriptor hTableDesc = createTableDescription(tableName, families);
        admin.createTable(hTableDesc);
    }

    /**
     * 创建表，并设置表的regions数量
     *
     * @param tableName  表描述
     * @param families   列族列表
     * @param startKey   键值起始值
     * @param endKey     键值终止值
     * @param numRegions 分区数量
     */
    public void createTable(String tableName, String[] families, byte[] startKey, byte[] endKey, int numRegions) throws IOException {
        HTableDescriptor hTableDesc = createTableDescription(tableName, families);
        admin.createTable(hTableDesc, startKey, endKey, numRegions);
    }

    /**
     * 创建表，并设置表的初始划分
     *
     * @param tableName 表名
     * @param families  列族列表
     * @param splitKeys 拆分的key值列表
     */
    public void createTable(String tableName, String[] families, String[] splitKeys) throws IOException {
        HTableDescriptor hTableDesc = createTableDescription(tableName, families);
        byte[][] splitKeysb = new byte[splitKeys.length][];
        for (int i = 0; i < splitKeys.length; i++) {
            splitKeysb[i] = Bytes.toBytes(splitKeys[i]);
        }
        admin.createTable(hTableDesc, splitKeysb);
    }

    /**
     * 标志一张表不可用
     *
     * @param table 表名
     */
    public void disableTable(String table) throws IOException {
        disableTable(TableName.valueOf(table));
    }

    /**
     * 标志一张表不可用
     */
    public void disableTable(TableName table) throws IOException {
        admin.disableTable(table);
    }

    /**
     * 删除一张表
     */
    public void dropTable(String table) throws IOException {
        dropTable(TableName.valueOf(table));
    }

    /**
     * 删除一张表
     */
    public void dropTable(TableName table) throws IOException {
        if (admin.tableExists(table)) {
            if (admin.isTableEnabled(table))
                admin.disableTable(table);
            admin.deleteTable(table);
        }
    }

    /**
     * 列出所有表
     */
    public HTableDescriptor[] listTables() throws IOException {
        return admin.listTables();
    }

    /**
     * 向表中插入某一列的值
     *
     * @param table 表名
     * @param row   行键
     * @param fam   列簇
     * @param qual  列修饰符
     * @param value 值
     */
    public void put(String table, String row, String fam, String qual, String value) {
        try (Table tb1 = connection.getTable(TableName.valueOf(table))) {
            Put put = new Put(Bytes.toBytes(row));
            put.addColumn(Bytes.toBytes(fam), Bytes.toBytes(qual), Bytes.toBytes(value));
            tb1.put(put);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 导入多条数据
     *
     * @param table  表名
     * @param row    行键
     * @param fam    列簇
     * @param quals  列修饰符
     * @param values 值
     */
    public void put(String table, String row, String fam, String[] quals, String[] values) {
        if (quals.length != values.length) {
            throw new IllegalArgumentException("数组长度不一致");
        }
        try (Table tb = connection.getTable(TableName.valueOf(table))) {
            List<Put> puts = new ArrayList<>();
            Put put = new Put(Bytes.toBytes(row));
            int i = 0;
            for (String qual : quals) {
                put.addColumn(Bytes.toBytes(fam), Bytes.toBytes(qual), Bytes.toBytes(values[i]));
                i++;
            }
            puts.add(put);
            tb.put(puts);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 批量入库
     */
    public void put(String table, List<Put> puts) {
        try (Table tb = connection.getTable(TableName.valueOf(table))) {
            tb.put(puts);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 创建一个Put对象
     *
     * @param row    rowkey
     * @param fam    列簇名称
     * @param quals  列名称
     * @param values 值
     */
    public Put buildPut(String row, String fam, String[] quals, String[] values) {
        if (quals.length != values.length)
            throw new IllegalArgumentException("数组长度不一致");
        Put put = new Put(Bytes.toBytes(row));
        for (int i = 0; i < quals.length; i++)
            put.addColumn(Bytes.toBytes(fam), Bytes.toBytes(quals[i]), Bytes.toBytes(values[i]));
        return put;
    }

    /**
     * 创建表的描述
     *
     * @param table    表名
     * @param families 列族名
     * @return 表的描述对象
     */
    private HTableDescriptor createTableDescription(String table, String... families) throws IOException {
        TableName tableName = TableName.valueOf(table);
        if (admin.tableExists(tableName)) {
            throw new TableExistsException();
        } else {
            HTableDescriptor hTableDesc = new HTableDescriptor(tableName);
            if (families != null)
                for (String family : families) {
                    HColumnDescriptor hColumnDesc = new HColumnDescriptor(family);
                    hTableDesc.addFamily(hColumnDesc);
                }
            return hTableDesc;
        }
    }

    /**
     * 根据某一列的值获取记录
     */
    public ResultScanner getScanner(String table, String fam, String col, String value) {
        try (Table tb1 = connection.getTable(TableName.valueOf(table))) {
            Scan scan = new Scan();
            scan.setCaching(2000);
            scan.setMaxVersions();
            SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes(fam), Bytes.toBytes(col), CompareFilter.CompareOp.EQUAL, Bytes.toBytes(value));
            scan.setFilter(filter);
            return tb1.getScanner(scan);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取表中所有的数据
     */
    public ResultScanner getScanner(String table) {
        try (Table tb = connection.getTable(TableName.valueOf(table))) {
            Scan scan = new Scan();
            scan.setCaching(2000);
            scan.setMaxVersions();
            return tb.getScanner(scan);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 根据表名与行键获取记录
     */
    public Result get(String table, String rowKey) {
        try (Table tb = connection.getTable(TableName.valueOf(table))) {
            Get get = new Get(Bytes.toBytes(rowKey));
            return tb.get(get);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static Dataset<Row> resultScannerToDataset(ResultScanner scanner) throws IOException {
        List<edu.zju.gis.dldsj.type.Row> rows = new ArrayList<>();
        StructType structType = new StructType();
        Result result = scanner.next();
        if (result != null) {
            result.listCells().stream().map(cell -> new StructField(Bytes.toString(CellUtil.cloneFamily(cell)) + "." + Bytes.toString(CellUtil.cloneQualifier(cell)),
                    DataType.STRING)).forEach(structType::addField);
        }
        while (result != null) {
            List<Cell> cells = result.listCells();
            cells.stream().map(cell -> new StructField(Bytes.toString(CellUtil.cloneFamily(cell)) + "." + Bytes.toString(CellUtil.cloneQualifier(cell)),
                    DataType.STRING)).forEach(structType::addField);
            String[] s = cells.stream().map(cell -> Bytes.toString(CellUtil.cloneValue(cell))).toArray(String[]::new);
            rows.add(new edu.zju.gis.dldsj.type.Row(s));
            s = result.listCells().stream().map(cell -> Bytes.toString(CellUtil.cloneValue(cell))).toArray(String[]::new);
            rows.add(new edu.zju.gis.dldsj.type.Row(s));
            result = scanner.next();
        }
        return new Dataset<>(structType, rows);
    }

    /**
     * 根据列修饰符将结果转换为Dataset（当存在非String类型列时，会出现乱码）
     *
     * @param results HBase查询结果
     * @see HBaseHelper#resultToModel(Class, Result...)
     */
    public static Dataset<edu.zju.gis.dldsj.type.Row> resultToDataset(Result... results) {
        List<edu.zju.gis.dldsj.type.Row> rows = new ArrayList<>();
        StructType structType = new StructType();
        if (results != null && results.length > 0) {
            results[0].listCells().stream().map(cell -> new StructField(Bytes.toString(CellUtil.cloneFamily(cell)) + "." + Bytes.toString(CellUtil.cloneQualifier(cell)),
                    DataType.STRING)).forEach(structType::addField);
            for (Result result : results) {
                List<Cell> cells = result.listCells();
                cells.stream().map(cell -> new StructField(Bytes.toString(CellUtil.cloneFamily(cell)) + "." + Bytes.toString(CellUtil.cloneQualifier(cell)),
                        DataType.STRING)).forEach(structType::addField);
                String[] s = cells.stream().map(cell -> Bytes.toString(CellUtil.cloneValue(cell))).toArray(String[]::new);
                rows.add(new edu.zju.gis.dldsj.type.Row(s));
                s = result.listCells().stream().map(cell -> Bytes.toString(CellUtil.cloneValue(cell))).toArray(String[]::new);
                rows.add(new edu.zju.gis.dldsj.type.Row(s));
            }
        }
        return new Dataset<>(structType, rows);
    }

    /**
     * 根据列修饰符将结果转换为模型对象
     *
     * @param clazz   结果模型类
     * @param results HBase查询结果
     * @param <T>     结果模型类
     */
    public static <T> List<T> resultToModel(Class<T> clazz, Result... results) throws IllegalAccessException, InstantiationException {
        if (results == null || results.length == 0)
            return Collections.emptyList();
        List<T> models = new ArrayList<>();
        List<Field> fields = Arrays.asList(clazz.getDeclaredFields());
        List<String> fieldNames = new ArrayList<>();
        List<Class> fieldTypes = new ArrayList<>();
        for (Field field : fields) {
            fieldNames.add(field.getName());
            fieldTypes.add(field.getType());
            field.setAccessible(true);
        }
        for (Result result : results) {
            T model = clazz.newInstance();
            models.add(model);
            for (Cell cell : result.listCells()) {
                String qual = Bytes.toString(CellUtil.cloneQualifier(cell));
                int index = fieldNames.indexOf(qual);
                if (index < 0)
                    continue;
                Object value;
                byte[] bytes = CellUtil.cloneValue(cell);
                switch (fieldTypes.get(index).getName()) {
                    case "short":
                    case "java.lang.Short":
                        value = Bytes.toShort(bytes);
                        break;
                    case "int":
                    case "java.lang.Integer":
                        value = Bytes.toInt(bytes);
                        break;
                    case "long":
                    case "java.lang.Long":
                        value = Bytes.toLong(bytes);
                        break;
                    case "float":
                    case "java.lang.Float":
                        value = Bytes.toFloat(bytes);
                        break;
                    case "double":
                    case "java.lang.Double":
                    case "java.math.BigDecimal":
                        value = Bytes.toDouble(bytes);
                        break;
                    case "boolean":
                    case "java.lang.Boolean":
                        value = Bytes.toBoolean(bytes);
                        break;
                    case "java.lang.String":
                        value = Bytes.toString(bytes);
                        break;
                    default:
                        value = Bytes.toString(bytes);
                }
                fields.get(index).set(model, value);
            }
        }
        return models;
    }

    /**
     * 获取表中所有的数据
     */
    public ResultScanner getScanner(String table, int offset, long size) {
        try (Table tb = connection.getTable(TableName.valueOf(table))) {
            Scan scan = new Scan().setRowOffsetPerColumnFamily(offset).setMaxResultSize(size).setCaching(2000).setMaxVersions();
            return tb.getScanner(scan);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
